package com.zp.datax.service.impl;

import com.zp.datax.core.DataxJobProviderFactory;
import com.zp.datax.core.DataxJobStatus;
import com.zp.datax.core.model.DataxJobInfo;
import com.zp.datax.core.model.DataxJobTrigger;
import com.zp.datax.core.result.ReturnT;
import com.zp.datax.core.util.JacksonUtil;
import com.zp.datax.core.util.ThreadPoolFactory;
import com.zp.datax.dao.DataxJobInfoDao;
import com.zp.datax.service.DataxJobInfoService;
import lombok.NonNull;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.File;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class DataxJobInfoServiceImpl implements DataxJobInfoService {

    @Resource
    private DataxJobInfoDao dataxJobInfoDao;
    @Autowired
    private DataxJobProviderFactory dataxJobProviderFactory;

    private final static Set<String> PARAMETERS;

    static {
        PARAMETERS = new HashSet<>(4);
        PARAMETERS.add("startId");
        PARAMETERS.add("endId");
        PARAMETERS.add("startTime");
        PARAMETERS.add("endTime");
    }

    private String getDataxHome() {
        return dataxJobProviderFactory.getDataxHome();
    }

    private String getJobDir() {
        return String.format(
                "%s%sjob"
                , getDataxHome()
                , File.separator);
    }

    private String getJobPath(String filename) {
        return String.format(
                "%s%s%s.json"
                , getJobDir()
                , File.separator
                , filename);
    }

    @Override
    public Map<String, Object> pageList(int start, int length, String name) {

        List<DataxJobInfo> pageList = dataxJobInfoDao.pageList(start, length, name);
        pageList = pageList.stream()
                .map(info -> {

                    //敏感数据使用*替换
                    String encryption = encryption(info.getContent());
                    info.setContent(encryption);

                    String params = info.getParams();
                    if (Objects.nonNull(params)
                            && !"".equals(params)) {
                        String[] array = params.split(",");
                        params = Arrays.stream(array)
                                .map(s -> s.concat(String.format("=请输入%s的值", s)))
                                .collect(Collectors.joining(","));
                        info.setParams(params.substring(0, params.length() - 1));
                    }
                    return info;
                })
                .collect(Collectors.toList());

        int pageListCount = dataxJobInfoDao.pageListCount(start, length, name);

        // package result
        Map<String, Object> maps = new HashMap<String, Object>();
        maps.put("recordsTotal", pageListCount);        // 总记录数
        maps.put("recordsFiltered", pageListCount);    // 过滤后的总记录数
        maps.put("data", pageList);                    // 分页列表

        return maps;
    }

    /**
     * 根据文件名删除文件
     *
     * @param sysId
     * @return
     */
    @Override
    public ReturnT<String> remove(final Integer sysId) {

        DataxJobInfo exists_dataxJobInfo = dataxJobInfoDao.load(sysId);
        if (Objects.isNull(exists_dataxJobInfo)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "任务不存在");
        }

        if (exists_dataxJobInfo.getStatus() == DataxJobStatus.NORMAL.getValue()) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "任务正在运行中，请先暂停");
        }

        dataxJobInfoDao.remove(sysId);
        //todo 删除执行计划
        dataxJobProviderFactory.getDataxJobThread().removeDataxJobTrigger(sysId);

        return ReturnT.SUCCESS;
    }

    /**
     * 参数配对
     *
     * @param params
     * @return
     */
    private boolean matchParams(final String params) {
        boolean flag = true;
        if (Objects.nonNull(params)
                && !"".equals(params)) {

            if (params.endsWith(",")) {
                flag = false;
            } else {

                flag = Arrays.stream(params.trim().split("[,]"))
                        .map(String::trim)
                        .allMatch(PARAMETERS::contains);
            }
        }

        return flag;
    }

    /**
     * 更新文件
     *
     * @param dataxJobInfo
     * @return
     */
    @Override
    public ReturnT<String> update(DataxJobInfo dataxJobInfo) {

        DataxJobInfo exists_dataxJobInfo = dataxJobInfoDao.load(dataxJobInfo.getSysId());
        if (Objects.isNull(exists_dataxJobInfo)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "任务不存在");
        }

        final String params = dataxJobInfo.getParams();
        if (!matchParams(params)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "参数名仅限[startId|endId|startTime|endTime]并按‘,’分隔");
        }

        if (exists_dataxJobInfo.getStatus() == DataxJobStatus.NORMAL.getValue()) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "任务正在运行中，请先暂停");
        }

        //检验json内容是否有变化，没有变化不更新，减少运维人员每次编辑时都要手动重置json内容
        String content = dataxJobInfo.getContent();
        String encryption = encryption(exists_dataxJobInfo.getContent());
        if (!Objects.equals(encryption, content)) {//有变化才更新
            //更新job文件
            writerJob(dataxJobInfo.getName(), content);
        } else {//未有变化不更新
            dataxJobInfo.setContent(null);
        }

        dataxJobInfo.setModifyTime(LocalDateTime.now().toString());
        dataxJobInfoDao.update(dataxJobInfo);


        DataxJobTrigger dataxJobTrigger = DataxJobTrigger.builder()
                .taskId(exists_dataxJobInfo.getSysId())
                .cron(dataxJobInfo.getCron())
                .params(params)
                .range(dataxJobInfo.getRange())
                .unit(dataxJobInfo.getUnit())
                .build();

        dataxJobProviderFactory.getDataxJobThread().updateDataxJobTrigger(dataxJobTrigger);

        return ReturnT.SUCCESS;

    }

    /**
     * 将用户名和密码使用*替换
     *
     * @param src
     * @return
     */
    private String encryption(String src) {

        String[] arrays = src.split("[\\,\\{\\}\\[\\]]");
        for (String str : arrays) {
            if (str.indexOf("username") != -1) {
                src = src.replace(str.trim(), "\"username\": \"******\"");
                continue;
            }
            if (str.indexOf("password") != -1) {
                src = src.replace(str.trim(), "\"password\": \"******\"");
                continue;
            }
        }

        return src;
    }

    /**
     * 生成json文件
     *
     * @param filename
     * @param content
     */
    @SneakyThrows
    private String writerJob(String filename, String content) {

        String path = getJobPath(filename);

        JacksonUtil.writeValue((w, v) -> w.write(Objects.toString(v)), new File(path), content);

        return path;

    }

    /**
     * 新增文件
     *
     * @param dataxJobInfo
     * @return
     */
    @Override
    public ReturnT<String> add(DataxJobInfo dataxJobInfo) {

        DataxJobInfo exists_dataxJobInfo = dataxJobInfoDao.findByName(dataxJobInfo.getName());
        if (Objects.nonNull(exists_dataxJobInfo)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "任务已存在");
        }

        final String params = dataxJobInfo.getParams();
        if (!matchParams(params)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "参数名仅限[startId|endId|startTime|endTime]并按‘,’分隔");
        }

        //生成job文件
        String path = writerJob(dataxJobInfo.getName(), dataxJobInfo.getContent());
        String now = LocalDateTime.now().toString();
        dataxJobInfo.setCreateTime(now);
        dataxJobInfo.setModifyTime(now);
        dataxJobInfo.setPath(path);
        dataxJobInfoDao.save(dataxJobInfo);

        //任务id
        Integer sysId = dataxJobInfo.getSysId();

        DataxJobTrigger dataxJobTrigger = DataxJobTrigger.builder()
                .name(dataxJobInfo.getName())
                .taskId(sysId)
                .path(path)
                .params(params)
                .cron(dataxJobInfo.getCron())
                .range(dataxJobInfo.getRange())
                .unit(dataxJobInfo.getUnit())
                .build();

        dataxJobProviderFactory.getDataxJobThread().addDataxJobTrigger(dataxJobTrigger);

        return ReturnT.SUCCESS;
    }

    @SneakyThrows
    @Override
    public ReturnT<String> exec(@NonNull Integer sysId, String params) {

        ThreadPoolFactory.INSTANCE.getThreadPool().execute(() -> dataxJobProviderFactory.getDataxJobThread().pageExec(sysId, params));

        return new ReturnT<>(ReturnT.SUCCESS_CODE, "调用成功，处理中...");
    }

    @Override
    public List<DataxJobInfo> searchAll() {
        return dataxJobInfoDao.findAll();
    }

    @Override
    public ReturnT<String> stop(Integer sysId) {

        DataxJobInfo exists_dataxJobInfo = dataxJobInfoDao.load(sysId);
        if (Objects.isNull(exists_dataxJobInfo)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "任务不存在");
        }

        DataxJobInfo dataxJobInfo = DataxJobInfo.builder()
                .sysId(sysId)
                .status(DataxJobStatus.STOP.getValue())
                .modifyTime(LocalDateTime.now().toString())
                .build();

        dataxJobInfoDao.update(dataxJobInfo);
        //todo 删除执行计划
        dataxJobProviderFactory.getDataxJobThread().remove(sysId);

        return ReturnT.SUCCESS;
    }

    @Override
    public ReturnT<String> start(Integer sysId) {

        DataxJobInfo exists_dataxJobInfo = dataxJobInfoDao.load(sysId);
        if (Objects.isNull(exists_dataxJobInfo)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "任务不存在");
        }

        DataxJobInfo dataxJobInfo = DataxJobInfo.builder()
                .sysId(sysId)
                .status(DataxJobStatus.NORMAL.getValue())
                .modifyTime(LocalDateTime.now().toString())
                .build();

        dataxJobInfoDao.update(dataxJobInfo);
        //todo 添加执行计划
        dataxJobProviderFactory.getDataxJobThread().push(sysId);

        return ReturnT.SUCCESS;
    }


}
